package cn.doitedu.sql;

import beans.UserAction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class _15_LookupJoinPrinciple {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);


        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("action-log")
                .setGroupId("gxx")
                .setClientIdPrefix("cxx")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();


        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "kfk");

        stream.map(json -> JSON.parseObject(json, UserAction.class))
                .keyBy(UserAction::getUid)
                .process(new KeyedProcessFunction<Integer, UserAction, String>() {

                    PreparedStatement preparedStatement;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        Connection connection = DriverManager.getConnection("jdbc:mysql://doitedu:3306/doit46","root","root");
                        preparedStatement = connection.prepareStatement("select * from student where id = ?");

                    }

                    @Override
                    public void processElement(UserAction userAction, KeyedProcessFunction<Integer, UserAction, String>.Context ctx, Collector<String> out) throws Exception {

                        int uid = userAction.getUid();
                        preparedStatement.setInt(1,uid);

                        ResultSet resultSet = preparedStatement.executeQuery();

                        resultSet.next();
                        String name = resultSet.getString("name");
                        String gender = resultSet.getString("gender");
                        int age = resultSet.getInt("age");
                        float salary = resultSet.getFloat("salary");

                        out.collect(userAction + " ==> " + name+","+age+","+gender+","+salary);
                    }
                }).print();

        env.execute();

    }

}
